#ifndef __M_SERVER_H__
#define __M_SERVER_H__

#include<iostream>
#include<vector>
#include<algorithm>
#include<cassert>
#include<cstring>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/eventfd.h>
#include <thread>
#include <mutex>
#include<condition_variable> 
#include <unistd.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <functional>
#include <sys/epoll.h>
#include <unordered_map>
#include <memory>
#include <sys/timerfd.h>
#include<sys/select.h>
#include<ctime>
#include <typeinfo>
#include<any>
#include<signal.h>
using namespace std;
#define BUFFER_SIZE 1024
class Buffer
{
    private:
    
    vector<char> _buffer;
    uint64_t _reader_idx;//读指针
    uint64_t _writer_idx;//写指针

    public:
    Buffer():_reader_idx(0),_writer_idx(0),_buffer(BUFFER_SIZE){}
    char* Begin(){return &*_buffer.begin();}//迭代器指针解引用为地址然后取地址返回
    char *WPosition(){return Begin()+_writer_idx;}//获取写入初始地址
    char*RPosition(){return Begin()+_reader_idx;}//获取读取初始地址
    uint64_t TailSize(){
       // cout<<_buffer.size()-_writer_idx<<endl;
        return _buffer.size()-_writer_idx;}//末尾空间大小-写偏移之后的空间大小
    uint64_t HeadSize(){return _reader_idx;}//开头空间大小-读偏移之前的空间大小
    uint64_t Readablesize(){return _writer_idx-_reader_idx;}//写入但还没读的大小
    void MoveReadOffset(uint64_t len){
        if(len==0) return ;
        assert(len<=Readablesize());
      _reader_idx=  _reader_idx+len;
    //  cout<<"radix:"<<_reader_idx<<endl;
    }//读指针可以向后移动一段距离
    void MoveWriteOffset(uint64_t len){
        assert(len<=TailSize());
      //  cout<<"?????????"<<endl;
       _writer_idx= _writer_idx+len;
      //  cout<<_writer_idx<<endl;
    }//写指针可以向后移动一段距离
    void EnsureWriteSpace(uint64_t len)//保证有足够的空间写入数据
    {
        if(TailSize()>=len){return ;}
        else if(TailSize()+HeadSize()>=len)//末尾空间大小加上开头空间大小装得下的话把先前的部分挪到前面
        {
            uint64_t hnr= Readablesize();
            copy(RPosition(),RPosition()+hnr,Begin());
            _reader_idx=0;
            _writer_idx=hnr;
        }else{//空间不够就扩容
        _buffer.resize(_writer_idx+len);
        }
    }

    void Write(const void*data,uint64_t len)//把数据写入buffer
    {
        //有足够空间，拷贝数据进去
        if(len==0) return ;//没数据
        
        EnsureWriteSpace(len);//保证有足够空间
        const char* d=(const char*)data;//类型匹配
        copy(d,d+len,WPosition());

    }
    void Writeandpush(const void*data,uint64_t len)
    {
        Write(data,len);
        MoveWriteOffset(len);
    }

    void WriteString(const string&data)
    {
        Write(data.c_str(),data.size());//data.c_str//()类型是const char* 
    }

    void  WriteStringandPush(const string&data)
    {
        Write(data.c_str(),data.size());//data.c_str//()类型是const char* 
     //   cout<<data<<": "<<data.size()<<endl;
         MoveWriteOffset(data.size());
      //   cout<<Readablesize()<<endl;
    }

    void WriteBuffer(Buffer&data)
    {
        Write(data.RPosition(),data.Readablesize());//把参数的已经写入但还没读的数据写入当前缓冲区中
    }

     void WriteBufferandpush(Buffer&data)
     {
        Write(data.RPosition(),data.Readablesize());//把参数的已经写入但还没读的数据写入当前缓冲区中
        MoveWriteOffset(data.Readablesize());
     }

     void Read(void*buff,uint64_t len)//把已经写入但还没读的数据读出
     {
        if(len==0) return ;
        assert(len<=Readablesize());
        //cout<<"::::::::::"<<endl;
        copy(RPosition(),RPosition()+len,(char*)buff); //把已经写入但还没读的数据拷贝到buff中
    //    cout<<"+++++++++++"<<endl;
     }

     void ReadandPop(void* buff,uint64_t len)
     {
        Read(buff,len);
        MoveReadOffset(len);
     //   cout<<"<<<<<<<<<<"<<endl;
     }
     string ReadAsstring(uint64_t len)//读一行
     {
        assert(Readablesize()>=len);
       std:: string strs;
        strs.resize(len);
        Read(&strs[0],len);//str[0]类型是const char*
        return strs;
     }

       std:: string ReadAsstringandPop(uint64_t len)//读一行
        {
        std::string tmp=  ReadAsstring(len);
            MoveReadOffset(len);//移动读指针
            return tmp;
        }

     char*FindCRLF()//找换行符
     {
        char*res=(char*)memchr((void*)RPosition(),'\n',Readablesize());//在已经写入但还没读的数据读到换行符的位置
        //第一个参数是要查找的内存地址，第二个参数是要查找的字符，第三个参数是要查找的长度（区间）
        return res;//如果没有换行符则返回null
     }
     string Getline()
     {
        char*pos=FindCRLF();
        if(pos==NULL) return "";
        return ReadAsstring(pos-RPosition()+1);//+1是为了把\n读出
     }
     string Getlineandpop()
     {
        string str=Getline();
        MoveReadOffset(str.size());
        return str;
     }

     void clear()//清空缓冲区
     {
        _reader_idx=0;
        _writer_idx=0;
     }
};

#define MAX_LISTEN 1024
#define IPPROTO_TCp 0
#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL ERR
//\是续航符time是时间函数，strftime把时间格式化为一个时间字符串;如果传递的level大于LOG_LEVEL就打印
#define LOG(level,format,...) {\
    if(level>=LOG_LEVEL) {\
    time_t t=time(NULL);\
    struct tm*ltm=localtime(&t);\
    char tmp[32]={0};\
    strftime(tmp,31,"%H:%M:%S",ltm);\
fprintf(stdout,"[%p %s %s:%d] " format "\n",(void*)pthread_self(),tmp,__FILE__,__LINE__,##__VA_ARGS__);\
}\
}
//stdout是FILE*类型表示输出的文件流，"%s:%d"是格式化字符串，__FILE__和__LINE__是预定义的宏，表示当前文件名和行号
#define INT_LOG(format,...) LOG(INF,format,##__VA_ARGS__)
#define DBG_LOG(format,...) LOG(DBG,format,##__VA_ARGS__)
#define ERR_LOG(format,...) LOG(ERR,format,##__VA_ARGS__)
//#define MAX_LISTEN 1024
 class   NetWork{

 public:
 NetWork(){
     DBG_LOG("Network initialized");
    signal(SIGPIPE,SIG_IGN);
//忽略SIGPIPE信号---当client端和server端通讯断开时，server端还在给client端send数据时会抛异常会导致server端进程退出
//
    //如果不忽略SIGPIPE信号，server端会抛出SIGPIPE异常，导致程序崩溃
    //如果忽略SIGPIPE信号，server端会返回EPIPE错误码，表示管道破裂
    //可以通过捕获EPIPE错误码来处理这种情况
    
 }

 };

class Socket{

private:
int _sockfd;

public:
Socket():_sockfd(-1){}
Socket(int fd):_sockfd(fd){}
~Socket(){Close();}
int Getfd(){return _sockfd;}
void Close()
{
    if(_sockfd!=-1)
    {
        close(_sockfd);
        _sockfd=-1;
    }
}

bool Create()
{
    _sockfd=socket(AF_INET,SOCK_STREAM,IPPROTO_TCp);//返回的是一个socket文件描述符
    if(_sockfd<0)
    {
        ERR_LOG("create socket failed");
        return false;
    }
    return true; 
}
bool Bind(const std::string &ip,uint16_t port)//绑定地址和端口
{
    struct sockaddr_in addr;
    addr.sin_family=AF_INET;
    addr.sin_port=htons(port);// htons把主机字节序转换为网络字节序
    addr.sin_addr.s_addr=inet_addr(ip.c_str());//inet_addr把字符串IP地址转换为网络字节序的IP地址
    if(bind(_sockfd,(struct sockaddr*)&addr,sizeof(addr))<0)
    {
        ERR_LOG("bind socket failed");
        return false;
    }
    return true;
}
bool Listen(int backlog=MAX_LISTEN)
{
    if(listen(_sockfd,backlog)<0)
    {
        ERR_LOG("listen socket failed");
        return false;
    }
    return true;
}
bool Connect(const std::string &ip,uint16_t port)
{
    struct sockaddr_in addr;
    addr.sin_family=AF_INET;
    addr.sin_port=htons(port);
    addr.sin_addr.s_addr=inet_addr(ip.c_str());
    if(connect(_sockfd,(struct sockaddr*)&addr,sizeof(addr))<0)
    {
        ERR_LOG("connect socket failed");
        return false;
    }
    return true;
}

int Accept()
{

    int newfd=accept(_sockfd,NULL,NULL);//接受连接
    if(newfd<0)
    {
        ERR_LOG("accept socket failed");
        return -1;
    }
    return newfd;
}
///接受数据
ssize_t Rev(void*buff,size_t len,int flag=0)
{
  //  std::cout << "_sockfd" << _sockfd << std::endl;
    ssize_t ret=recv(_sockfd,buff,len,flag);
    //EAGAIN 当前socket的接收缓冲区中没有数据了，在⾮阻塞的情况下才会有这
    //EINTR 表⽰当前socket的阻塞等待，被信号打断了，
    if(ret<=0)
    {
        if(errno==EAGAIN||errno==EINTR)
        {
            return 0;
        }
        else
        {
            perror("recv failed: ");
            ERR_LOG("recv data failed");
            return -1;
        }
    }
    return ret;
}

ssize_t NonBlockRecv(void* buf,size_t len)//MSG_DONTWAIT表示当前接收为非阻塞
{

    return Rev(buf,len,MSG_DONTWAIT);
}
//发送数据
ssize_t Send(const void* buf,size_t len,int flag=0)//flag默认为0
{
    
    ssize_t ret=send(_sockfd,buf,len,flag);//往sockfd发送数据
    if(ret<0)
    {
        if(errno==EAGAIN||errno==EINTR)
        {
            return 0;
        }
        else
        {
            ERR_LOG("send data failed");
            return -1;
        }
    }
    return ret;//返回实际发送的字节数
}
ssize_t NonBlockSend(const void* buf,size_t len)
{
    if(len==0) return 0;
    return Send(buf,len,MSG_DONTWAIT);// MSG_DONTWAIT 表⽰当前发送为⾮阻塞
}

bool CreateServer(uint16_t port,const std::string &ip="0.0.0.0",bool block_flag=false)
{
    //1. 创建套接字，2. 绑定地址，3. 开始监听，4. 设置⾮阻塞， 5. 启动地址重⽤
    if(!Create())
    {
        return false;
    }
    
    if(!Bind(ip,port))
    {
        return false;
    }
      if(!Listen())
    {
        return false;
    }
    if(block_flag)NonBlock();//设置为非阻塞
  ReuseAddress();//设置套接字选项---开启地址端⼝重⽤ 
    return true;
}

bool CreateClient(uint16_t port,const std::string &ip,bool block_flag=false)
{
    if(!Create())
    {
        return false;
    }
    if(block_flag)NonBlock();//设置为非阻塞
    ReuseAddress();//设置套接字选项---开启地址端⼝重⽤
    if(!Connect(ip,port))
    {
        return false;
    }
    return true;
}   

void ReuseAddress()
{
    int val=1;
    setsockopt(_sockfd,SOL_SOCKET,SO_REUSEADDR,(void*)&val,sizeof(int));//允许重用本地地址和端口
    val=1;
    setsockopt(_sockfd,SOL_SOCKET,SO_REUSEPORT,(void*)&val,sizeof(int));//许多个套接字同时绑定到同一个IP地址和端口号上，而不会导致绑定冲突
    
}

void NonBlock(){
    int flag=fcntl(_sockfd,F_GETFL,0);//使用F_GETFL命令可以获取或设置文件的状态标志
    fcntl(_sockfd,F_SETFL,flag|O_NONBLOCK);//吧当前套接字设置为非阻塞
}

};

class Poller;
class EventLoop;
class Channel{
private:
uint32_t _events;//当前需要监控的事件
uint32_t _revents;//当前链接触发的事件
EventLoop *_EventLoop;//当前链接所属的eventloop
int _fd;//当前链接的文件描述符
using EventCallback=std::function<void()>;//回调函数
Poller *_poller;//poller指针
EventCallback _read_callback;//读事件回调函数
EventCallback _write_callback;//写事件回调函数
EventCallback _error_callback;//错误事件回调函数
EventCallback _close_callback;//关闭事件回调函数
EventCallback _event_callback;//事件回调函数

public:
Channel(EventLoop*evp, int fd):_fd(fd),_events(0),_revents(0),_EventLoop(evp){}
~Channel(){}
int Fd(){return _fd;}
uint32_t Events(){return _events;}
void SetRevents(uint32_t events){_revents=events;}//设置就绪事件
void SetReadCallback(const EventCallback &cb){_read_callback=cb;}
void SetWriteCallback(const EventCallback &cb){_write_callback=cb;}
void SetErrorCallback(const EventCallback &cb){_error_callback=cb;}
void SetCloseCallback(const EventCallback &cb){_close_callback=cb;}
void SetEventCallback(const EventCallback &cb){_event_callback=cb;}

bool ReadAble(){return _revents&EPOLLIN;}//判断是否有读事件
bool WriteAble(){return _revents&EPOLLOUT;}//判断是否有写事件

void EnableRead(bool flag=true){_events|=EPOLLIN;Update();}//设置读事件
void EnableWrite(bool flag=true){_events|=EPOLLOUT;Update();}//设置写事件

void DisableRead(){_events&=~EPOLLIN;Update();}//取消读事件
void DisableWrite(){_events&=~EPOLLOUT;Update();}//取消写事件

void DisableAll(){_events=0;Update();}//取消所有事件
void Remove();//移除事件后面调用eventloop 
void Update();//更新事件
void HandleEvent(){
    if((_revents&EPOLLHUP)||(_revents&EPOLLIN)||(_revents&EPOLLPRI))//对方关闭连接,有数据可读，有紧急数据可读
    { if(_event_callback)_event_callback();//不管什么事件都调用这个函数
        if(_read_callback)_read_callback();//
        
    }
    if(_revents&EPOLLOUT)//有数据可写
    { if(_event_callback)_event_callback();//不管什么事件都调用这个函数
        if(_write_callback)_write_callback();
        
    }
    else if(_revents&EPOLLERR)//错误事件
    {
        if(_error_callback)_error_callback();
    }
   else if(_revents&EPOLLRDHUP)//对方关闭连接
    {
        if(_close_callback)_close_callback();
    }
   
    
}//事件处理函数


};

#define MAX_EPOLLEVENTS 1024 
class Poller{   
private:
int _epfd;
struct epoll_event _events[MAX_EPOLLEVENTS];//保存活跃事件的数组
std::unordered_map<int,Channel*> _channels;
private:
void update(Channel*channel,int op)
{
    int fd=channel->Fd();//
    struct epoll_event ev;
    ev.data.fd=fd;
    ev.events=channel->Events();
   int ret= epoll_ctl(_epfd,op,fd,&ev);//该函数作用是修改epoll的事件监控
   if(ret<0){
    ERR_LOG("epoll_ctl failed");
    abort();
   }
   return;
}
//判断channel是否添加了监控(添加进hash表)
bool hasChannel(Channel*channel)
{
    auto it=_channels.find(channel->Fd());
    return it!=_channels.end();
}
public:
Poller()
{
    _epfd=epoll_create(MAX_EPOLLEVENTS);//创建一个epoll实例，返回一个文件描述符
    if(_epfd<0)
    {
        ERR_LOG("epoll_create failed");
        abort();
    }
}

//添加或修改channel
void updateEvent(Channel*Channel)
{
    if(!hasChannel(Channel))//真-在->更新
    {//不在就添加
        _channels[Channel->Fd()]=Channel;//添加进_channels hash表---第一次到这里把
        update(Channel,EPOLL_CTL_ADD);//添加到epoll中
        DBG_LOG("add channel fd:%d",Channel->Fd());
    }
    else
    {
        update(Channel,EPOLL_CTL_MOD);
    }
}

void RemoveEvent(Channel*channel)//移除channel出hash表
{
    if(!hasChannel(channel)) return ;
    update(channel,EPOLL_CTL_DEL);
    _channels.erase(channel->Fd());
}

void Poll(std::vector<Channel*>*active)
{
    DBG_LOG("before epoll_wait");
  //  cout<<"befoe epoll_wait"<<endl;
    int nfd=epoll_wait(_epfd,_events,MAX_EPOLLEVENTS,-1);
    //函数的返回值表示就绪的事件数量--第一次接收到即监听的客户端发来了连接
    //阻塞等待活跃事件第一个是监控的fd，第二个是事件数组的指针，第三个是事件的个数，第四个是等待的事件-1表示阻塞等待
 //   cout<<"after epoll_wait"<<endl;
    if(nfd<0)
{
    if(errno==EINTR) return ;//阻塞被信号打断
    ERR_LOG("epoll_wait failed");
    abort();    
}
for(int i=0;i<nfd;i++)//遍历活跃事件，尾插进vcetor active
{
   
    auto it=_channels.find(_events[i].data.fd);//如果fd就绪就必然存在于_channels中
    assert(it!=_channels.end());
    it->second->SetRevents(_events[i].events);//
    active->push_back(it->second);//添加进vector active活跃事件集中
    //这里是什么意思？？？
}
return;
}
};


using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
class TimeTask{
    private:
    uint64_t _id;
    uint32_t _timeout;
    bool _canceled;//false表示没有被取消 
    TaskFunc _task_cb;
    ReleaseFunc _retask;
    public:
    TimeTask( uint64_t id,uint32_t timeout, const TaskFunc &cb):_id(id),_timeout(timeout),_task_cb(cb){}
    //构造
    ~TimeTask(){_task_cb();_retask();}//析构
    void SetRelease(const  ReleaseFunc&cb ){_retask=cb;}//释放
    uint32_t getdelaytime()
    {
        return _timeout;
    }
    void Cancel(){_canceled=true;}
   
    uint32_t Delaytime(){return _timeout;}
};

class EventLoop;
class TimerWheel
{
    private:
    int _tick;
    int _capacity;
    using weaktask=std::weak_ptr<TimeTask>;
    using PtrTask=std::shared_ptr<TimeTask>;
    vector<vector<PtrTask>> _wheel;//时间轮
    unordered_map<uint64_t,weaktask> _timers;
    //定时器id和weakptr的映射--该成员由于是fd的id和weakptr的关系，
    //有可能多个线程通过同一个fd操作同一个weakptr，会造成线程安全问题---67节20分大概
    //timers会添加或者移除定时器即使得weakptr的引用计数发生变化，即线程安全问题
    EventLoop* _loop;
    int _timerfd;//可读事件回调就是读取计时器，执行任务
    std::unique_ptr<Channel> _timer_channel;//在析构时主动释放
    private:
 void RemoveTimer(uint64_t id)//移除定时器
    {
        auto it=_timers.find(id);
        if(it!=_timers.end())
        {
            _timers.erase(id);
        }
    }
    static int CreateTimerfd()
    {
        int timerfd=timerfd_create(CLOCK_MONOTONIC,0);
        if(timerfd<0)
        {
            ERR_LOG("timerfd_create failed");
            abort();
        }  
    
    struct itimerspec itime;
//第一次超时间隔为1秒
itime.it_value.tv_sec=1;
itime.it_value.tv_nsec=0;
//第一次往后的超时间隔为1秒
itime.it_interval.tv_sec=1;
itime.it_interval.tv_nsec=0;

 timerfd_settime(timerfd,0,&itime,NULL);//启动计时器
 return timerfd;
   }

   int ReadTimefd()
   {
    uint64_t times;
    //有可能处理其他描述符时的事件花费事件太长，因此在处理定时器描述符时发送超时很多次，
    //read的距离上一次read定时器超时次数
    int ret=read(_timerfd,&times,sizeof(times));
    if(ret<0)
    {
        if(errno==EAGAIN||errno==EINTR) return 0;
        ERR_LOG("read timerfd failed");
        abort();
    }
    return times;
   }
   void RunTimerTask()//每秒钟走一次
   {
       _tick=(_tick+1)%_capacity;
       _wheel[_tick].clear();//tick数组那个位置，对应位置的sharedptr就要释放
   }

   void Ontime()
   {
    int times=ReadTimefd();//取出超时多少次-取出来后_times清空
    for(int i=0;i<times;i++)
    {
        RunTimerTask();//超时多少次执行多少次
    }
   }

   void TimerAddInloop(uint64_t id,uint32_t delay,const TaskFunc& cb)
   {
         PtrTask pt(new TimeTask(id,delay,cb));//new个sharedptr
         pt->SetRelease(bind( &TimerWheel::RemoveTimer,this,id));//TimeTask的SetRelease函数绑定成RemoveTimer函数
         int pos=(_tick+delay)%_capacity;
         _wheel[pos].push_back(pt);//sharedptr添加进轮子
         _timers[id]=weaktask(pt);//timers里面放的是weakptr不会增加count供后面刷新使用
   }

   

   void TimerRefreshInloop(uint64_t id)
   {
         auto it=_timers.find(id);
         if(it==_timers.end())
         {
              return ;
         }
         PtrTask pt=it->second.lock();
         //用_timers的second对应weakptr构造sharedptr，不增加count
         //放进轮子
         int delay=pt->getdelaytime();
         int pos=(_tick+delay)%_capacity;
          _wheel[pos].push_back(pt);//添加进轮子
   }

   void TimerCancelInloop(uint64_t id)
   {
    auto it=_timers.find(id);
    if(it==_timers.end())
    {
        return ;
    }
    PtrTask pt=it->second.lock();
    if(pt)
    {
        pt->Cancel();
    }
   }
    public:
    
    TimerWheel(EventLoop*loop):_tick(0),_capacity(60),_wheel(_capacity),_loop(loop),_timerfd(CreateTimerfd()),_timer_channel(new Channel(_loop,_timerfd)){
        _timer_channel->SetReadCallback(std::bind(&TimerWheel::Ontime,this));
        //每秒钟超时，即每秒钟调用一次读事件回调，即调用Ontime函数
        _timer_channel->EnableRead();//启动读监控
    }


    

    void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb);
    //要求在eventloop线程中调用，因此在共有成员函数中调用私有成员函数解决线程安全问题

    void TimeaRefrash(uint64_t id);//刷新
    
    void TimerCancel(uint64_t id);
    bool HasTimer(uint64_t id)//这里如果是被外面调用可能会产生线程安全问题
    {
        return _timers.find(id)!=_timers.end();
    }
};



class EventLoop{
private:
using Functor=std::function<void()>;
std::thread::id _thread_id;
int _event_fd;//用于唤醒IO事件可能引起的队列任务阻塞
Poller _poller;
std::vector<Functor> _task;//任务队列
std::mutex _mutex;//互斥锁
std::unique_ptr<Channel> _event_channel; // 添加 _event_channel 成员
TimerWheel _timerwheel;
public:
void RunAllTask(){//执行线程池的所有任务
    std::vector<Functor> functor;
    {
        std::unique_lock<std::mutex> _lock(_mutex);//加锁，让后面的task内的任务交换到functor中是线程安全的
        _task.swap(functor);//task中的任务交换到functor中
    }//这里的_lock会在离开作用域时自动释放锁
    for(auto &t:functor)//执行functor中的任务
    {
        t();
    }
}
static int CreateEventFd( ) {
    int evtfd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (evtfd < 0) {
        ERR_LOG("Failed to create eventfd");
        abort();
    }
    return evtfd;
}
static void ReadEventFd(int fd)
{
    uint64_t one=0;
    int n= read(fd,&one,sizeof(one));//读取eventfd
    if(n<0)
    {
        if(errno==EAGAIN||errno==EINTR) return ;
        ERR_LOG("read eventfd failed");
        abort();
    }
    return;
}

void WeakUpEventFd()//函数作用是？
{
    uint64_t one=1;
    int n=write(_event_fd,&one,sizeof(one));//写入eventfd
    if(n<0)
    {
        if(errno==EAGAIN||errno==EINTR) return ;
        ERR_LOG("write eventfd failed");
        abort();
    }
    return;
}

public:
EventLoop():_thread_id(std::this_thread::get_id())
,_event_fd(CreateEventFd())
,_event_channel(new Channel(this, _event_fd))
,_timerwheel(this)
{ //给eventfd设置读回调，读取通知次数
    _event_channel->SetReadCallback(std::bind(ReadEventFd,_event_fd));
    _event_channel->EnableRead();
}  


void start()//事件监控-》就绪事件处理-》执行任务
{
    while(1){
    std::vector<Channel*> active;
    DBG_LOG("std::vector<Channel*> active;");
   
   _poller.Poll(&active);
   DBG_LOG(" _poller.Poll(&active);");

   for(auto &it:active)
   {
     DBG_LOG(" before HandleEvent");
       it->HandleEvent();
        // DBG_LOG("有一个事件");
      
   }
    RunAllTask();//执行活跃事件执行完后执行任务队列
    DBG_LOG(" RunAllTask()");
}
}
bool IsInLoopThread(){return _thread_id==std::this_thread::get_id();}//判断当前线程是否是对应eventloop的线程
void RuninLoop(const Functor&cb)//判断要执行的任务是否在当前线程，如果是就直接执行，不是就加入任务队列
{
    if(IsInLoopThread())
    {
        cb();
    }
    else
    {
        QueueInLoop(cb);
    }
}
void QueueInLoop(const Functor&cb)//任务加入线程池
{
    {
        std::unique_lock<std::mutex> _lock(_mutex);
        _task.push_back(cb);
    }
}
void UpdateEvent(Channel*channel){_poller.updateEvent(channel);}//添加或修改描述符的事件监控
void RemoveEvent(Channel*channel){_poller.RemoveEvent(channel);}//移除描述符的事件监控
void AssertInLoopThread(){assert(IsInLoopThread());}//断言当前线程是否是对应eventloop的线程
void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb){_timerwheel.TimerAdd(id,delay,cb);}//添加定时器
void TimerRefresh(uint64_t id){_timerwheel.TimeaRefrash(id);}//刷新定时器
void TimerCancel(uint64_t id){_timerwheel.TimerCancel(id);}//取消定时器
bool HasTimer(uint64_t id){return _timerwheel.HasTimer(id);}//判断是否有定时器
};

class Any{
    private:
    class holder{//父类
        public:
        //虚函数=0表示子类必须重写（实例化看对象）
        virtual ~holder(){}
        virtual  const std::type_info& type()=0;//返回数据类型
        virtual holder* clone()=0;//父类对象指针
    };
    template<class T>
    class placeholder:public holder{//子类
    public:
    placeholder(const T&val):_val(val){}//构造
     
        virtual const std::type_info& type(){return typeid(T);}//返回子类数据类型
        virtual holder* clone(){return  new placeholder(_val);}//根据当前对象克隆出一个新的子类对象

        public:
        T _val;
    };
    holder* _content;//父类指针
    public:
    Any():_content(NULL){}
    template<class T>
    Any(const T&val):_content(new placeholder<T>(val)){}//用val构造一个子类对象
    Any(const Any&other):_content( other._content?other._content->clone():NULL){}//拷贝构造：用Any对象构造--返回一个父类指针，指向子类对象
    ~Any(){delete _content;}

   Any &swap(Any&other)
   {
   std:: swap(_content,other._content);
    return *this; 
   }

   template<class T>
   T*get()//获取数据地址
   {
    assert(typeid(T)==_content->type());
    return &((placeholder<T>*)_content)->_val;
    //holder是父类，父类没有val，因此强转为子类，_content是个地址加*，返回的是val的地址
   }
   template<class T>
   Any& operator=(const T&val)//赋值重载用val构造
   {
   Any(val).swap(*this);//用val构造一个Any对象，然后交换当前对象和新对象
    return *this;
   }

   Any &operator=(Any&other)//赋值重载用other对象构造
   {
   other.swap(*this);//交换当前对象和other对象
    return *this;
   }
   


};

class Connection;
typedef enum{DISCONNECTED,CONNECING,CONNECTED,DISCONNECTING} ConnState;
//DISCONNECTED已经关闭连接,CONNECING准备好连接待处理,
//CONNECTED准备好连接且可以通信,DISCONNECTING等待处理关闭连接
using PtrConnection=std::shared_ptr<Connection>;//防止被随便释放
class Connection:public std::enable_shared_from_this<Connection>
//它提供了获取指向当前对象的 shared_ptr 和 weak_ptr 的功能
{
private:
uint64_t _conn_id;//唯一id
int _sockfd;//文件描述符
bool _enable_inactive_release;//是否允许非活跃连接释放,
EventLoop* _loop;//事件循环
Channel _channel;//事件通道
ConnState _state;//连接状态
Buffer _input_buffer;//输入缓冲区
Buffer _output_buffer;//输出缓冲区
Socket _socket;//套接字管理
Any _context;//上下文

using ConnectionCallback=std::function<void(const PtrConnection&)>;//连接回调函数
using CloseCallback=std::function<void(const PtrConnection&)>;//关闭回调函数
using MessageCallback=std::function<void(const PtrConnection&,Buffer*)>;//消息回调函数
using AnyEventCallback=std::function<void(const PtrConnection&)>;//事件回调函数

//using WriteCompleteCallback=std::function<void(const PtrConnection&)>;//写完成回调函数
//using EventCallback=std::function<void(const PtrConnection&)>;//事件回调函数
ConnectionCallback _connection_callback;//连接回调函数
MessageCallback _message_callback;//消息回调函数
//WriteCompleteCallback _write_complete_callback;//写完成回调函数
CloseCallback _close_callback;//关闭回调函数
CloseCallback _server_close_callback;//服务器端关闭回调函数--组件内的关闭回调
//EventCallback _event_callback;//事件回调函数
AnyEventCallback _event_callback;//事件回调函数
private:
void HandleRead()
{//1.接收socket数据，放到缓冲区

    char buff[65536];
    ssize_t ret=_socket.NonBlockRecv(buff,65535);//表示非阻塞接收数据
    if(ret<0)
    {
        return ShutdownInloop();//如果接收数据失败，不能直接关闭连接还能判断现在接收缓冲区是否还有数据
    }
    
    _input_buffer.Writeandpush(buff,ret);
    if(_input_buffer.Readablesize()>0)//接收缓冲区内有数据调用回调函数
    {
       return _message_callback(shared_from_this(),&_input_buffer);
    }
}

void HandleWrite()
{
    ssize_t ret=_socket.NonBlockSend(_output_buffer.RPosition(),_output_buffer.Readablesize());
    if(ret<0)
    {
        if(_input_buffer.Readablesize()>0)//如果输入缓冲区还有数据就要处理
        {
        _message_callback(shared_from_this(), &_input_buffer);
        }
        return ReleaseInloop();//关闭释放
    }
    _output_buffer.MoveReadOffset(ret);//有数据就读inb->outb，读完移动指针
    if(_output_buffer.Readablesize()==0)//输出缓冲区没有数据待发送
    {
        _channel.DisableWrite();//关闭写事件
        if(_state==DISCONNECTING)//如果是关闭连接状态
        {
          return   ReleaseInloop();//关闭连接
        }
    }
}

void HandleClose()
{//连接挂断，套接字瘫痪，检查缓冲区，若有数据就处理，然后断开连接
    if(_input_buffer.Readablesize()>0)
    {
        _message_callback(shared_from_this(), &_input_buffer);
    }
  return   ReleaseInloop();
}

void HandleError()
{
    ERR_LOG("HandleError");
    HandleClose();
}
void HandleEvent()//刷新定时器，推迟定时任务，调用函数
{
    if(_enable_inactive_release==true)//如果允许非活跃连接释放
    {_loop->TimerRefresh(_conn_id);}//刷新定时器
    if(_event_callback)_event_callback(shared_from_this());
}
void EstablishedInloop()//修改连接状态，启动读事件，调用回调函数
{//修改为已经连接状态
    assert(_state==CONNECING);//当前状态必须是半连接状态
    _state=CONNECTED;//修改状态
    DBG_LOG("before EnableRead");
   
    _channel.EnableRead();//启动读事件  
    DBG_LOG("after EnableRead");
   
    if(_connection_callback)_connection_callback(shared_from_this());//调用连接回调函数

    DBG_LOG("after _connection_callback");
}

void ReleaseInloop()//实际的释放接口
{//修改连接状态
    _state==DISCONNECTED;
    _channel.Remove();//移除事件监控
    //关闭描述符
    _socket.Close();
    //定时器队列中还有任务则取消任务
    if(_loop->HasTimer(_conn_id))CancelInactiveReleaseInloop();
    //调用关闭回调函数，先关闭用户再关闭服务器
    if(_close_callback)_close_callback(shared_from_this());
    if(_server_close_callback)_server_close_callback(shared_from_this());
}

void SendInloop(Buffer&buff){//把数据放到输出缓冲区，打开写监控
    if(_state==DISCONNECTED) return ;
    _output_buffer.WriteBufferandpush(buff);
    if(_channel.WriteAble()==false)
    {
        _channel.EnableWrite();
    }
}

void ShutdownInloop()//不是关闭连接，而是判断缓冲区是否还有数据
{
    _state=DISCONNECTING;//调成半关闭状态
    if(_input_buffer.Readablesize()>0)//如果输入缓冲区还有数据就要处理
    {
        _message_callback(shared_from_this(),&_input_buffer);
    }
    if(_output_buffer.Readablesize()>0)//如果输出缓冲区还有数据没有写出去则开启写事件
    {
        if(_channel.WriteAble()==false)
        {
            _channel.EnableWrite();
        }
    }
    if(_output_buffer.Readablesize()==0)//输出缓冲区没有数据
    {
        ReleaseInloop();//关闭连接
    }
}
//启动非活跃连接超时机制
void EnableInactiveReleaseInloop(uint32_t sec)//启动非活跃连接释放
{
    DBG_LOG("EnableInactiveReleaseInloop sec:%d",sec);
    _enable_inactive_release=true;
    if(_loop->HasTimer(_conn_id))//如果有定时器就刷新
    {
        _loop->TimerRefresh(_conn_id);
    }
   //如果不存在销毁任务就新建
   else  _loop->TimerAdd(_conn_id,sec,std::bind(&Connection::ReleaseInloop,this));//if执行了刷新了就不要再添加了
}


void CancelInactiveReleaseInloop(){//直接取消非活跃连接释放

_enable_inactive_release=false;
if(_loop->HasTimer(_conn_id))
{
    _loop->TimerCancel(_conn_id);
}

}
    

void UpgradeInloop(const Any&context//更新上下文
,const ConnectionCallback&connection_cb  
,const MessageCallback&message_cb
,const AnyEventCallback&event_vb
,const CloseCallback&close_cb)
{
    _context=context;
    _connection_callback=connection_cb;
    _message_callback=message_cb;
    _event_callback=event_vb;
    _close_callback=close_cb;
}

public:
Connection(EventLoop*loop,uint64_t connid,int fd):_conn_id(connid),
_sockfd(fd),  
_enable_inactive_release(false),
_loop(loop),
_channel(loop,fd),
//_socket(fd),
_state(CONNECING){
    DBG_LOG("Set channel any function %d",_socket);
   
    _channel.SetReadCallback(std::bind(&Connection::HandleRead,this));
    _channel.SetWriteCallback(std::bind(&Connection::HandleWrite,this));
    _channel.SetCloseCallback(std::bind(&Connection::HandleClose,this));
    _channel.SetErrorCallback(std::bind(&Connection::HandleError,this));
    _channel.SetEventCallback(std::bind(&Connection::HandleEvent,this));
}
~Connection(){
    ERR_LOG("release Connection: %p",this);
}
int Fd(){return _sockfd;}
int Id(){return _conn_id;}
bool Connected(){return _state==CONNECTED;}
void SetContext(const Any&context){_context=context;}
Any*GetContext(){return &_context;}
void SetConnectedCallback(const ConnectionCallback&cb){_connection_callback=cb;}
void SetMessageCallback(const MessageCallback&cb){_message_callback=cb;}
void SetCloseCallback(const CloseCallback&cb){_close_callback=cb;}
void SetServerCloseCallback(const CloseCallback&cb){_server_close_callback=cb;}
void SetAnyCallback(const AnyEventCallback&cb){_event_callback=cb;}
void Send(const char *data,size_t len){
   //data可能是个临时空间，如果直接将data加入任务队列，可能在执行该任务之前data已经被释放，因此用buff来过度
    Buffer buff;
    buff.Writeandpush(data,len);
    _loop->RuninLoop(std::bind(&Connection::SendInloop,this,buff));
}
void Established(){_loop->RuninLoop(std::bind(&Connection::EstablishedInloop,this));}
//提供给用户的接口，并不直接关闭，而是判断是否有数据要处理
void Shutdown(){_loop->RuninLoop(std::bind(&Connection::ShutdownInloop,this));}
//直接释放
void Release(){_loop->RuninLoop(std::bind(&Connection::ReleaseInloop,this));}
void EnableInactiveRelease(uint32_t sec){_loop->RuninLoop(std::bind(&Connection::EnableInactiveReleaseInloop,this,sec));}
//直接取消非活跃连接释放
void CancelInactiveRelease(){_loop->RuninLoop(std::bind(&Connection::CancelInactiveReleaseInloop,this));}
//该更新函数可能会导致线程安全，因此需要保证是在该线程中执行
void Upgrate(const Any&context//更新上下文
    ,const ConnectionCallback&connection_cb  
    ,const MessageCallback&message_cb
    ,const AnyEventCallback&event_vb
    ,const CloseCallback&close_cb)
    {
        _loop->AssertInLoopThread();//验证是否是在该线程执行
        _loop->RuninLoop(std::bind(&Connection::UpgradeInloop, this,context,connection_cb,message_cb,event_vb,close_cb));
        
    }

};

class LoopThread{
    private:
    std::mutex _mutex;//互斥锁
    std::condition_variable _cond;//条件变量
    EventLoop* _loop;//事件循环
    std::thread _thread;//线程

    private:
    void ThreadEntry()//线程函数
    {
        EventLoop loop;//创建事件循环---在这里创建保证loop绑定该当前线程
        {
            std::unique_lock<std::mutex> _lock(_mutex);//加锁
            _loop=&loop;//将事件循环指针赋值给_loop
            _cond.notify_all();//通知等待的线程
        }
        loop.start();//开始事件循环
        
    }

    public:
    LoopThread():_loop(NULL),_thread(std::bind(&LoopThread::ThreadEntry,this))//线程函数
    {}
    EventLoop* GetLoop()//获取事件循环
    {
        DBG_LOG("LoopThread::GetLoop() begin");
        EventLoop* loop=nullptr;
        {
        std::unique_lock<std::mutex> _lock(_mutex);//加锁
        _cond.wait(_lock,[&](){return _loop!=NULL;});
        loop=_loop;//等待条件变量满足
        }
        return loop;//返回事件循环 
    }

};

class LoopThreadPool{//事件循环线程池   
private:
int thread_count;//从线程个数
int _next_idx;//下一个线程索引
EventLoop* _main_loop;//主事件循环
std::vector<LoopThread*> _threads;//线程池
std::vector<EventLoop*> _loops;//事件循环池

public:
LoopThreadPool(EventLoop*main_loop):thread_count(0),_next_idx(0),_main_loop(main_loop){}
void SetThreadCount(int count)//设置线程个数
{  thread_count=count;}

void Start()//启动线程池
{
    if(thread_count>0){
        _threads.resize(thread_count);//调整线程池大小
        _loops.resize(thread_count);//调整事件循环池大小
        for(int i=0;i<thread_count;i++)
        {
            _threads[i]=new LoopThread();//创建线程是使用互斥锁和条件变量的，因此还没new完之前是阻塞的
            _loops[i]=_threads[i]->GetLoop();//获取事件循环
        }
    }
    return;
}

EventLoop* GetNextLoop()//获取下一个事件循环
{
    if(thread_count==0) return _main_loop;//如果线程池为空就返回主事件循环
    
    _next_idx=(_next_idx+1)%_loops.size();//更新下一个索引--先1后0
    return _loops[_next_idx];//返回下一个事件循环
}

};

class Acceptor{//接受连接类
private:
Socket _socket;//套接字
EventLoop* _loop;//事件循环
Channel _channel;//事件通道
using AcceptorCallback=std::function<void(int)>;//接受连接回调函数
AcceptorCallback _acceptor_callback;//接受连接回调函数--第一次进来这里是Tcpserver的Newconnection函数

private:
void HandleRead()
{
    int newfd=_socket.Accept();//接受连接
    if(newfd<0)
    {
        ERR_LOG("accept failed");
        return;
    }
    if(_acceptor_callback)_acceptor_callback(newfd);//调用接受连接回调函数--调用Tcpserver的NewConnection函数
}


int CreateServer(int port)
{
    bool ret=_socket.CreateServer(port);//创建服务器套接字
    assert(ret==true);
    return _socket.Getfd();//返回套接字文件描述符
}

public:

//不能把读事件监控放到构造函数中，因为AcceptorCallback可能还没有设置好，
//此时如果有连接到来就会导致空指针异常，因此在AcceptorCallback设置好后再调用Listen函数
//AcceptorCallback是一个函数指针，指向接受连接的回调函数
Acceptor(EventLoop*loop,int port):_socket(CreateServer(port)),_loop(loop),_channel(loop, _socket.Getfd())
{
    _channel.SetReadCallback(std::bind(&Acceptor::HandleRead,this));
}

void SetAcceptorCallback(const AcceptorCallback&cb){_acceptor_callback=cb;}//设置接受连接回调函数

void LiSten()//开始监听
{
    _channel.EnableRead();//设置读事件
}

};

class Tcpserver{

private:
uint64_t _nextid;//自动增长的连接id
int _port;//服务器端口
int _timeout;//非活跃连接超时时间
bool _enable_inactive_release;//是否允许非活跃连接释放
EventLoop _baseloop;//事件循环
Acceptor _acceptor;//接受连接类
LoopThreadPool _threadpool;//事件循环线程池
std::unordered_map<uint64_t,PtrConnection> _connections;//连接池，存储

using ConnectionCallback=std::function<void(const PtrConnection&)>;//连接回调函数
using CloseCallback=std::function<void(const PtrConnection&)>;//关闭回调函数
using MessageCallback=std::function<void(const PtrConnection&,Buffer*)>;//消息回调函数
using AnyEventCallback=std::function<void(const PtrConnection&)>;//事件回调函数
using Functor=std::function<void()>;//任务函数
ConnectionCallback _connection_callback;//连接回调函数
MessageCallback _message_callback;//消息回调函数
CloseCallback _close_callback;//关闭回调函数
AnyEventCallback _event_callback;//事件回调函数

private:

void RunAfterInloop(const Functor&cb,int delay)//在eventloop线程中执行任务
{
    _nextid++;//每次调用该函数都会增加id

    _baseloop.TimerAdd(_nextid,delay,cb);//添加定时器
}

void NewConnection(int fd)//新连接到来
{
    _nextid++;//每次调用该函数都会增加id
    DBG_LOG("NewConnection fd:%d nextid:%lu",fd,_nextid);
    PtrConnection conn(new Connection(_threadpool.GetNextLoop(),_nextid,fd));//创建连接对象
   
    conn->SetConnectedCallback(_connection_callback);//设置连接回调函数
    conn->SetMessageCallback(_message_callback);//设置消息回调函数
    conn->SetCloseCallback(_close_callback);//设置关闭回调函数
    conn->SetAnyCallback(_event_callback);//设置事件回调函数
    conn->SetServerCloseCallback(std::bind(&Tcpserver::RemoveConnection,this,std::placeholders::_1));//设置服务器端关闭回调函数
   if(_enable_inactive_release)conn->EnableInactiveRelease(_timeout);//如果允许非活跃连接释放就启动非活跃连接释放
    conn->Established();//建立连接
    _connections[_nextid]=conn;//将连接添加到连接池中
}

void RemoveConnectionInloop(const PtrConnection&conn)//移除连接
{
    int id=conn->Id();//获取连接id
    auto it=_connections.find(id);//在连接池中查找连接
    if(it!=_connections.end())//如果找到连接
    {
        _connections.erase(it);//从连接池中移除连接
    }
}

void RemoveConnection(const PtrConnection&conn)//移除连接
{
    _baseloop.RuninLoop(std::bind(&Tcpserver::RemoveConnectionInloop,this,conn));//在eventloop线程中执行移除连接
}

public:
Tcpserver(int port):_port(port),_nextid(0),_enable_inactive_release(false),_acceptor(&_baseloop,port),_threadpool(&_baseloop){
    _acceptor.SetAcceptorCallback(std::bind(&Tcpserver::NewConnection,this,std::placeholders::_1));
    _acceptor.LiSten();//开始监听
}

void SetThreadCount(int count)//设置线程池线程个数
{
    _threadpool.SetThreadCount(count);
}

void SetConnectedCallback(const ConnectionCallback&cb)//设置连接回调函数
{
    
    _connection_callback=cb;
    DBG_LOG("SetConnectedCallback");
}

void SetMessageCallback(const MessageCallback&cb)//设置消息回调函数
{
    _message_callback=cb;
    DBG_LOG("SetConnectedCallback");
}

    void SetCloseCallback(const CloseCallback&cb)//设置关闭回调函数
{
    _close_callback=cb;}

    void SetAnyEventCallback(const AnyEventCallback&cb)//设置事件回调函数
{
    _event_callback=cb; }

    void EnableInactiveRelease(int timeout)//启动非活跃连接释放
{
    _enable_inactive_release=true;
    _timeout=timeout;//设置非活跃连接超时时间
}


void RunAfter(const Functor&task,int delay)//在eventloop线程中执行任务
{
    _baseloop.RuninLoop(std::bind(&Tcpserver::RunAfterInloop,this,task,delay));
}

void Start()//启动服务器
{
    _threadpool.Start();
    _baseloop.start();//开始事件循环
}
};



void Channel::Update(){return _EventLoop->UpdateEvent(this);}//移除事件后面调用eventloop
void Channel::Remove(){return _EventLoop->RemoveEvent(this);}//更新事件
void TimerWheel:: TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb) 
{
    _loop->RuninLoop(std::bind(&TimerWheel::TimerAddInloop,this,id,delay,cb));//TimerAddInloop存在线程安全问题
}
void TimerWheel:: TimeaRefrash(uint64_t id)//刷新
{
    _loop->RuninLoop(std::bind(&TimerWheel::TimerRefreshInloop,this,id));
}
void TimerWheel:: TimerCancel(uint64_t id)
{
    _loop->RuninLoop(std::bind(&TimerWheel::TimerCancelInloop,this,id));
}



#endif


